package com.poetic.market.analysis.process;

import com.poetic.market.analysis.domain.MarketingCountView;
import com.poetic.market.analysis.domain.MarketingUserBehavior;
import lombok.val;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class MarketingCountByChannel extends ProcessWindowFunction<Tuple2<Tuple2<String, String>, Long>, MarketingCountView, Tuple2<String, String>, TimeWindow> {


    @Override
    public void process(Tuple2<String, String> key, Context context, Iterable<Tuple2<Tuple2<String, String>, Long>> elements, Collector<MarketingCountView> out) throws Exception {
        long start = context.window().getStart();
        long end = context.window().getEnd();
        String channel = key.f0;
        String behaviorType = key.f1;
        long count = elements.spliterator().estimateSize();

        SimpleDateFormat parse = new SimpleDateFormat("yyyy/MM/dd-HH:mm:ss");
        MarketingCountView view = MarketingCountView.builder()
                .windowStart(parse.format(new Date(start)))
                .windowEnd(parse.format(new Date(end)))
                .channel(channel)
                .behavior(behaviorType)
                .count(count)
                .build();
        out.collect(view);
    }
}
